package org.rency.mq.kafka.consumer.support.factory;

import org.rency.mq.kafka.core.KafkaConstant;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.kafka.listener.KafkaDataListener;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * ${todo}
 *
 * @author: Created by user_rcy@163.com on 2016/10/21.
 * @version: V1.0.0
 * @since JDK 1.8
 */
public class KafkaContainerProperties implements FactoryBean<ContainerProperties>{

    private String                        consumerTopic;

    private ThreadPoolTaskExecutor        taskExecutor;

    private KafkaDataListener             messageListener;

    private long                          consumerPollTimeout;

    public KafkaContainerProperties(String topicName,KafkaDataListener messageListener){
        this(topicName,null,messageListener);
    }

    public KafkaContainerProperties(String topicName,ThreadPoolTaskExecutor taskExecutor,KafkaDataListener messageListener){
        this.consumerTopic = topicName;
        this.taskExecutor = taskExecutor;
        this.messageListener = messageListener;
        this.consumerPollTimeout = KafkaConstant.DEFAULT_CONSUMER_POLL_TIMEOUT;
    }

    private ContainerProperties getContainerProperties(){
        ContainerProperties properties = new ContainerProperties(this.consumerTopic);
        properties.setConsumerTaskExecutor(this.taskExecutor);
        properties.setListenerTaskExecutor(this.taskExecutor);
        properties.setMessageListener(this.messageListener);
        properties.setPollTimeout(this.consumerPollTimeout);
        return properties;
    }

    @Override
    public ContainerProperties getObject() throws Exception {
        return this.getContainerProperties();
    }

    @Override
    public Class<?> getObjectType() {
        return ContainerProperties.class;
    }

    @Override
    public boolean isSingleton() {
        return false;
    }

    public void setConsumerPollTimeout(long consumerPollTimeout) {
        this.consumerPollTimeout = consumerPollTimeout;
    }
}